package com.zengjianjun.canal.demo.consumer;

import cn.hutool.json.JSONUtil;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.stream.Collectors;

/**
 * @author zengjianjun
 */
public interface CanalConsumerInterface {

    String HOST_NAME = "192.168.137.128";

    int PORT = 11111;

    default String getColumnValue(CanalEntry.Column column) {
        return column == null ? null : column.getValue();
    }

    default List<CanalEntry.Entry> filterInsertEntry(Message message) {
        return message.getEntries().stream().filter(data -> {
            try {
                CanalEntry.EntryType entryType = data.getEntryType();
                if (entryType == CanalEntry.EntryType.TRANSACTIONBEGIN || entryType == CanalEntry.EntryType.TRANSACTIONEND) {
                    // 过滤开启/关闭事务操作类型
                    return false;
                }
                CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(data.getStoreValue());
                // 获取操作类型：insert/update/delete类型
                CanalEntry.EventType eventType = rowChange.getEventType();
                if (eventType != CanalEntry.EventType.INSERT) {
                    // 非新增类型
                    return false;
                }
            } catch (Exception e) {
                System.out.println("message filter error: data = " + JSONUtil.toJsonStr(data));
                System.out.println("error message: " + e.getLocalizedMessage());
                return false;
            }
            return true;
        }).collect(Collectors.toList());
    }
}
